package com.lagou;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.*;

public class HbaseTest {

    public Connection conn;
    public Configuration conf;

    @Before
    public void init() throws IOException {
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum","linux121,linux122");
        conf.set("hbase.zookeeper.property.clientPort","2181");
        conn = ConnectionFactory.createConnection(conf);
    }

    @After
    public void release() throws IOException {
        if(conn != null){
            conn.close();
        }
    }

    @Test
    public void scane2() throws IOException {
        Table e2 = conn.getTable(TableName.valueOf("e2"));
        ResultScanner scanner = e2.getScanner(new Scan());
        for (Result result : scanner) {
            Cell[] cells = result.rawCells();
            for (Cell cell : cells) {
                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                String columnFamily = Bytes.toString(CellUtil.cloneFamily(cell));
                String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println(rowkey + "----" + columnFamily + "---" + column + "---" + value);
            }
        }
        e2.close();
    }

    public void addFriends(String u1, String u2) throws IOException {
        Table e2 = conn.getTable(TableName.valueOf("e2"));
        Put put1 = new Put(Bytes.toBytes(u1));
        Put put2 = new Put(Bytes.toBytes(u2));
        put1.addColumn(Bytes.toBytes("friends"),Bytes.toBytes(u2),Bytes.toBytes(u2));
        put2.addColumn(Bytes.toBytes("friends"),Bytes.toBytes(u1),Bytes.toBytes(u1));
        e2.put(Arrays.asList(put1, put2));
        e2.close();
    }



    @Test
    public void initData() throws IOException {
        Table e2 = conn.getTable(TableName.valueOf("e2"));
        addFriends("uid1","uid2");
        addFriends("uid1","uid3");
        addFriends("uid1","uid4");
        addFriends("uid2","uid3");
        e2.close();
        scane2();
    }

    public void deleteFriends(String u1, String u2) throws IOException {
        Table e2 = conn.getTable(TableName.valueOf("e2"));
        Delete delete = new Delete(Bytes.toBytes(u1));
        delete.addColumn(Bytes.toBytes("friends"),Bytes.toBytes(u2));
        e2.delete(delete);
        e2.close();
    }

    @Test
    public void tst() throws IOException {
        Table e2 = conn.getTable(TableName.valueOf("e2"));
        Delete delete = new Delete(Bytes.toBytes("u1"));
        delete.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("u2"));
        System.out.println(new String(delete.getRow()));
        NavigableMap<byte[], List<Cell>> familyCellMap = delete.getFamilyCellMap();
        familyCellMap.forEach((bytes, cells) -> {
            for (Cell cell : cells) {
                System.out.println(Bytes.toString(CellUtil.cloneFamily(cell)));
                System.out.println(Bytes.toString(CellUtil.cloneQualifier(cell)));
            }
        });
    }

    @Test
    public void test2() throws IOException {
        scane2();//扫描
        TableName tableName = TableName.valueOf("e2");
        Table e2 = conn.getTable(tableName);
        // load Coprocessor
        HTableDescriptor tableDescriptor = e2.getTableDescriptor();
        tableDescriptor.addCoprocessor("com.lagou.MyProcessor",
                new Path("hdfs://linux121:9000/E2/E2.jar"),10000,null);// read-only error
        Admin admin = conn.getAdmin();
        admin.modifyTable(tableName,tableDescriptor);
        // delete
        deleteFriends("uid2","uid3");
        scane2();
        //unload
        tableDescriptor.removeCoprocessor("com.lagou.MyProcessor");
        admin.close();
        e2.close();
    }
}
